package com.flink.hbase.statemanagement.managed;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * BroadcastState示例
 * 
 * 展示如何使用BroadcastState进行广播状态管理：
 * 1. 动态配置规则管理
 * 2. 实时风控规则更新
 * 3. 用户标签动态匹配
 * 4. 产品价格动态调整
 * 
 * @author Flink Team
 */
public class BroadcastStateExample {
    
    /**
     * 配置规则事件（广播流）
     */
    public static class ConfigRule {
        public String ruleId;
        public String ruleName;
        public String ruleType;
        public Map<String, Object> conditions;
        public String action;
        public long timestamp;
        
        public ConfigRule() {}
        
        public ConfigRule(String ruleId, String ruleName, String ruleType, 
                         Map<String, Object> conditions, String action, long timestamp) {
            this.ruleId = ruleId;
            this.ruleName = ruleName;
            this.ruleType = ruleType;
            this.conditions = conditions;
            this.action = action;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("ConfigRule{ruleId='%s', ruleName='%s', ruleType='%s', conditions=%s, action='%s', timestamp=%d}", 
                ruleId, ruleName, ruleType, conditions, action, timestamp);
        }
    }
    
    /**
     * 用户交易事件（主流）
     */
    public static class UserTransaction {
        public String userId;
        public String transactionId;
        public double amount;
        public String currency;
        public String location;
        public String deviceType;
        public long timestamp;
        
        public UserTransaction() {}
        
        public UserTransaction(String userId, String transactionId, double amount, 
                             String currency, String location, String deviceType, long timestamp) {
            this.userId = userId;
            this.transactionId = transactionId;
            this.amount = amount;
            this.currency = currency;
            this.location = location;
            this.deviceType = deviceType;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("UserTransaction{userId='%s', transactionId='%s', amount=%.2f, currency='%s', location='%s', deviceType='%s', timestamp=%d}", 
                userId, transactionId, amount, currency, location, deviceType, timestamp);
        }
    }
    
    /**
     * 用户行为事件（主流）
     */
    public static class UserBehavior {
        public String userId;
        public String behaviorType;
        public String page;
        public String product;
        public Map<String, Object> properties;
        public long timestamp;
        
        public UserBehavior() {}
        
        public UserBehavior(String userId, String behaviorType, String page, 
                           String product, Map<String, Object> properties, long timestamp) {
            this.userId = userId;
            this.behaviorType = behaviorType;
            this.page = page;
            this.product = product;
            this.properties = properties;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("UserBehavior{userId='%s', behaviorType='%s', page='%s', product='%s', properties=%s, timestamp=%d}", 
                userId, behaviorType, page, product, properties, timestamp);
        }
    }
    
    /**
     * 产品价格事件（广播流）
     */
    public static class ProductPrice {
        public String productId;
        public double price;
        public String currency;
        public String region;
        public long timestamp;
        
        public ProductPrice() {}
        
        public ProductPrice(String productId, double price, String currency, String region, long timestamp) {
            this.productId = productId;
            this.price = price;
            this.currency = currency;
            this.region = region;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("ProductPrice{productId='%s', price=%.2f, currency='%s', region='%s', timestamp=%d}", 
                productId, price, currency, region, timestamp);
        }
    }
    
    /**
     * 风控规则处理函数
     * 使用BroadcastState动态更新风控规则
     */
    public static class RiskControlProcessor extends BroadcastProcessFunction<UserTransaction, ConfigRule, Tuple3<String, String, String>> {
        
        // 风控规则广播状态描述符
        private static final MapStateDescriptor<String, ConfigRule> RULE_STATE_DESCRIPTOR = 
            new MapStateDescriptor<>("risk-rules", String.class, ConfigRule.class);
        
        @Override
        public void processElement(UserTransaction transaction, ReadOnlyContext ctx, 
                                 Collector<Tuple3<String, String, String>> out) throws Exception {
            
            // 获取广播状态
            ReadOnlyBroadcastState<String, ConfigRule> broadcastState = ctx.getBroadcastState(RULE_STATE_DESCRIPTOR);
            
            // 检查所有风控规则
            for (Map.Entry<String, ConfigRule> entry : broadcastState.immutableEntries()) {
                ConfigRule rule = entry.getValue();
                
                if ("RISK_CONTROL".equals(rule.ruleType)) {
                    // 应用风控规则
                    String riskResult = applyRiskRule(transaction, rule);
                    if (!"PASS".equals(riskResult)) {
                        out.collect(new Tuple3<>(transaction.transactionId, rule.ruleId, riskResult));
                        
                        System.out.println(String.format("⚠️ 风控告警: 交易 %s 触发规则 %s，结果: %s", 
                            transaction.transactionId, rule.ruleName, riskResult));
                    }
                }
            }
        }
        
        @Override
        public void processBroadcastElement(ConfigRule rule, Context ctx, 
                                          Collector<Tuple3<String, String, String>> out) throws Exception {
            
            // 获取广播状态
            BroadcastState<String, ConfigRule> broadcastState = ctx.getBroadcastState(RULE_STATE_DESCRIPTOR);
            
            // 更新规则
            broadcastState.put(rule.ruleId, rule);
            
            System.out.println("📢 广播规则更新: " + rule.ruleName);
        }
        
        /**
         * 应用风控规则
         */
        private String applyRiskRule(UserTransaction transaction, ConfigRule rule) {
            Map<String, Object> conditions = rule.conditions;
            
            // 检查金额限制
            if (conditions.containsKey("maxAmount")) {
                double maxAmount = (Double) conditions.get("maxAmount");
                if (transaction.amount > maxAmount) {
                    return "AMOUNT_EXCEEDED";
                }
            }
            
            // 检查地理位置限制
            if (conditions.containsKey("restrictedLocation")) {
                String restrictedLocation = (String) conditions.get("restrictedLocation");
                if (transaction.location.equals(restrictedLocation)) {
                    return "LOCATION_RESTRICTED";
                }
            }
            
            // 检查设备类型限制
            if (conditions.containsKey("restrictedDevice")) {
                String restrictedDevice = (String) conditions.get("restrictedDevice");
                if (transaction.deviceType.equals(restrictedDevice)) {
                    return "DEVICE_RESTRICTED";
                }
            }
            
            return "PASS";
        }
    }
    
    /**
     * 用户标签匹配处理函数
     * 使用BroadcastState动态更新用户标签规则
     */
    public static class UserTagMatcher extends BroadcastProcessFunction<UserBehavior, ConfigRule, Tuple3<String, String, String>> {
        
        // 用户标签规则广播状态描述符
        private static final MapStateDescriptor<String, ConfigRule> TAG_RULE_STATE_DESCRIPTOR = 
            new MapStateDescriptor<>("tag-rules", String.class, ConfigRule.class);
        
        @Override
        public void processElement(UserBehavior behavior, ReadOnlyContext ctx, 
                                 Collector<Tuple3<String, String, String>> out) throws Exception {
            
            // 获取广播状态
            ReadOnlyBroadcastState<String, ConfigRule> broadcastState = ctx.getBroadcastState(TAG_RULE_STATE_DESCRIPTOR);
            
            // 检查所有标签规则
            for (Map.Entry<String, ConfigRule> entry : broadcastState.immutableEntries()) {
                ConfigRule rule = entry.getValue();
                
                if ("USER_TAG".equals(rule.ruleType)) {
                    // 应用标签规则
                    String tag = applyTagRule(behavior, rule);
                    if (tag != null) {
                        out.collect(new Tuple3<>(behavior.userId, rule.ruleId, tag));
                        
                        System.out.println(String.format("🏷️ 用户标签: 用户 %s 匹配规则 %s，获得标签: %s", 
                            behavior.userId, rule.ruleName, tag));
                    }
                }
            }
        }
        
        @Override
        public void processBroadcastElement(ConfigRule rule, Context ctx, 
                                          Collector<Tuple3<String, String, String>> out) throws Exception {
            
            // 获取广播状态
            BroadcastState<String, ConfigRule> broadcastState = ctx.getBroadcastState(TAG_RULE_STATE_DESCRIPTOR);
            
            // 更新规则
            broadcastState.put(rule.ruleId, rule);
            
            System.out.println("📢 广播标签规则更新: " + rule.ruleName);
        }
        
        /**
         * 应用标签规则
         */
        private String applyTagRule(UserBehavior behavior, ConfigRule rule) {
            Map<String, Object> conditions = rule.conditions;
            
            // 检查行为类型
            if (conditions.containsKey("behaviorType")) {
                String expectedBehavior = (String) conditions.get("behaviorType");
                if (!behavior.behaviorType.equals(expectedBehavior)) {
                    return null;
                }
            }
            
            // 检查页面
            if (conditions.containsKey("page")) {
                String expectedPage = (String) conditions.get("page");
                if (!behavior.page.equals(expectedPage)) {
                    return null;
                }
            }
            
            // 检查产品
            if (conditions.containsKey("product")) {
                String expectedProduct = (String) conditions.get("product");
                if (!behavior.product.equals(expectedProduct)) {
                    return null;
                }
            }
            
            // 返回标签
            return rule.action;
        }
    }
    
    /**
     * 产品价格动态调整处理函数
     * 使用BroadcastState动态更新产品价格
     */
    public static class DynamicPriceAdjuster extends BroadcastProcessFunction<UserBehavior, ProductPrice, Tuple3<String, String, Double>> {
        
        // 产品价格广播状态描述符
        private static final MapStateDescriptor<String, ProductPrice> PRICE_STATE_DESCRIPTOR = 
            new MapStateDescriptor<>("product-prices", String.class, ProductPrice.class);
        
        @Override
        public void processElement(UserBehavior behavior, ReadOnlyContext ctx, 
                                 Collector<Tuple3<String, String, Double>> out) throws Exception {
            
            // 获取广播状态
            ReadOnlyBroadcastState<String, ProductPrice> broadcastState = ctx.getBroadcastState(PRICE_STATE_DESCRIPTOR);
            
            // 如果用户查看了产品，返回当前价格
            if ("view".equals(behavior.behaviorType) && behavior.product != null) {
                ProductPrice price = broadcastState.get(behavior.product);
                if (price != null) {
                    out.collect(new Tuple3<>(behavior.userId, behavior.product, price.price));
                    
                    System.out.println(String.format("💰 用户 %s 查看产品 %s，当前价格: %.2f", 
                        behavior.userId, behavior.product, price.price));
                }
            }
        }
        
        @Override
        public void processBroadcastElement(ProductPrice price, Context ctx, 
                                          Collector<Tuple3<String, String, Double>> out) throws Exception {
            
            // 获取广播状态
            BroadcastState<String, ProductPrice> broadcastState = ctx.getBroadcastState(PRICE_STATE_DESCRIPTOR);
            
            // 更新价格
            broadcastState.put(price.productId, price);
            
            System.out.println("📢 广播价格更新: " + price.productId + " -> " + price.price);
        }
    }
    
    /**
     * 模拟配置规则事件源
     */
    public static class MockConfigRuleSource implements SourceFunction<ConfigRule> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private int ruleIdCounter = 1;
        
        @Override
        public void run(SourceContext<ConfigRule> ctx) throws Exception {
            while (running) {
                // 生成风控规则
                if (random.nextBoolean()) {
                    Map<String, Object> conditions = new HashMap<>();
                    conditions.put("maxAmount", 1000.0 + random.nextDouble() * 9000.0);
                    conditions.put("restrictedLocation", random.nextBoolean() ? "RESTRICTED_AREA" : "ALLOWED_AREA");
                    conditions.put("restrictedDevice", random.nextBoolean() ? "mobile" : "desktop");
                    
                    ConfigRule rule = new ConfigRule(
                        "risk_rule_" + ruleIdCounter++,
                        "风控规则_" + ruleIdCounter,
                        "RISK_CONTROL",
                        conditions,
                        "BLOCK",
                        System.currentTimeMillis()
                    );
                    
                    ctx.collect(rule);
                } else {
                    // 生成标签规则
                    Map<String, Object> conditions = new HashMap<>();
                    conditions.put("behaviorType", "view");
                    conditions.put("page", "product");
                    conditions.put("product", "product" + (random.nextInt(3) + 1));
                    
                    ConfigRule rule = new ConfigRule(
                        "tag_rule_" + ruleIdCounter++,
                        "标签规则_" + ruleIdCounter,
                        "USER_TAG",
                        conditions,
                        "INTERESTED_USER",
                        System.currentTimeMillis()
                    );
                    
                    ctx.collect(rule);
                }
                
                Thread.sleep(10000 + random.nextInt(5000)); // 10-15秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟用户交易事件源
     */
    public static class MockUserTransactionSource implements SourceFunction<UserTransaction> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] currencies = {"USD", "EUR", "CNY"};
        private final String[] locations = {"US", "EU", "CN", "RESTRICTED_AREA"};
        private final String[] deviceTypes = {"mobile", "desktop", "tablet"};
        private int transactionIdCounter = 1;
        
        @Override
        public void run(SourceContext<UserTransaction> ctx) throws Exception {
            while (running) {
                UserTransaction transaction = new UserTransaction(
                    userIds[random.nextInt(userIds.length)],
                    "tx_" + (transactionIdCounter++),
                    100.0 + random.nextDouble() * 10000.0, // 100-10100随机金额
                    currencies[random.nextInt(currencies.length)],
                    locations[random.nextInt(locations.length)],
                    deviceTypes[random.nextInt(deviceTypes.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(transaction);
                Thread.sleep(2000 + random.nextInt(3000)); // 2-5秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟用户行为事件源
     */
    public static class MockUserBehaviorSource implements SourceFunction<UserBehavior> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] behaviorTypes = {"view", "click", "purchase", "share"};
        private final String[] pages = {"home", "product", "cart", "profile"};
        private final String[] products = {"product1", "product2", "product3", "product4", "product5"};
        
        @Override
        public void run(SourceContext<UserBehavior> ctx) throws Exception {
            while (running) {
                Map<String, Object> properties = new HashMap<>();
                properties.put("sessionId", "session_" + random.nextInt(1000));
                properties.put("referrer", "google");
                
                UserBehavior behavior = new UserBehavior(
                    userIds[random.nextInt(userIds.length)],
                    behaviorTypes[random.nextInt(behaviorTypes.length)],
                    pages[random.nextInt(pages.length)],
                    products[random.nextInt(products.length)],
                    properties,
                    System.currentTimeMillis()
                );
                
                ctx.collect(behavior);
                Thread.sleep(1000 + random.nextInt(2000)); // 1-3秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟产品价格事件源
     */
    public static class MockProductPriceSource implements SourceFunction<ProductPrice> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] productIds = {"product1", "product2", "product3", "product4", "product5"};
        private final String[] currencies = {"USD", "EUR", "CNY"};
        private final String[] regions = {"US", "EU", "CN"};
        
        @Override
        public void run(SourceContext<ProductPrice> ctx) throws Exception {
            while (running) {
                ProductPrice price = new ProductPrice(
                    productIds[random.nextInt(productIds.length)],
                    50.0 + random.nextDouble() * 450.0, // 50-500随机价格
                    currencies[random.nextInt(currencies.length)],
                    regions[random.nextInt(regions.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(price);
                Thread.sleep(8000 + random.nextInt(4000)); // 8-12秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 运行BroadcastState示例
     */
    public static void runBroadcastStateExample(StreamExecutionEnvironment env) throws Exception {
        System.out.println("🚀 开始运行BroadcastState示例...");
        
        // 1. 风控规则处理示例
        DataStream<ConfigRule> ruleStream = env.addSource(new MockConfigRuleSource())
            .name("配置规则流");
        
        DataStream<UserTransaction> transactionStream = env.addSource(new MockUserTransactionSource())
            .name("用户交易流");
        
        // 创建广播流
        BroadcastStream<ConfigRule> ruleBroadcastStream = ruleStream.broadcast(
            new MapStateDescriptor<>("risk-rules", String.class, ConfigRule.class)
        );
        
        DataStream<Tuple3<String, String, String>> riskResultStream = transactionStream
            .connect(ruleBroadcastStream)
            .process(new RiskControlProcessor())
            .name("风控规则处理");
        
        riskResultStream.print("风控结果");
        
        // 2. 用户标签匹配示例
        DataStream<UserBehavior> behaviorStream = env.addSource(new MockUserBehaviorSource())
            .name("用户行为流");
        
        BroadcastStream<ConfigRule> tagRuleBroadcastStream = ruleStream.broadcast(
            new MapStateDescriptor<>("tag-rules", String.class, ConfigRule.class)
        );
        
        DataStream<Tuple3<String, String, String>> tagResultStream = behaviorStream
            .connect(tagRuleBroadcastStream)
            .process(new UserTagMatcher())
            .name("用户标签匹配");
        
        tagResultStream.print("用户标签");
        
        // 3. 产品价格动态调整示例
        DataStream<ProductPrice> priceStream = env.addSource(new MockProductPriceSource())
            .name("产品价格流");
        
        BroadcastStream<ProductPrice> priceBroadcastStream = priceStream.broadcast(
            new MapStateDescriptor<>("product-prices", String.class, ProductPrice.class)
        );
        
        DataStream<Tuple3<String, String, Double>> priceResultStream = behaviorStream
            .connect(priceBroadcastStream)
            .process(new DynamicPriceAdjuster())
            .name("产品价格动态调整");
        
        priceResultStream.print("价格调整");
        
        System.out.println("✅ BroadcastState示例配置完成");
    }
    
    /**
     * 单独运行BroadcastState示例的主方法
     */
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 配置状态后端
        com.flink.hbase.statemanagement.config.StateBackendConfig
            .configureDevRocksDB(env);
        
        // 运行示例
        runBroadcastStateExample(env);
        
        // 执行作业
        env.execute("BroadcastState示例");
    }
} 